package com.showdor.springboot;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.sql.*;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TrafficDataFromDorisProducer {

    private static final String TOPIC_NAME = "jtts-traffice-topic";
    // Kafka 服务器地址
    private static final String BOOTSTRAP_SERVERS = "192.168.139.185:32092";
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final Producer<String, String> producer = new KafkaProducer<>(getKafkaProperties());
    private static final int ADD_DAY =  63;

    public static void main(String[] args) throws InterruptedException, ParseException {
        objectMapper.registerModule(new JavaTimeModule());  // 关键修复
        objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS); // 可选：禁用时间戳格式

// 连接到doris数据库，从2025-06-10 00:00:00开始，每10秒递增从表motor_vehicle_record中获取数据，按pass_time字段升序
        String dorisUrl = "jdbc:mysql://192.168.139.170:9030/jg_yldf_db?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai";
        //String dorisUrl="jdbc:mysql://192.168.139.185:3306/jg-yldf-db-tz?characterEncoding=UTF-8&useUnicode=true&useSSL=false&tinyInt1isBit=false&allowPublicKeyRetrieval=true&serverTimezone=Asia/Shanghai";
        String dorisUsername = "root";
        String dorisPassword = "htgx@1234";
        String dorisQuerySql = "select * from motor_vehicle_record where pass_time >= '%s' and pass_time<'%s' order by pass_time asc";
        String startTime = "2025-06-09 23:31:00";
        String endTime = "2025-06-09 23:31:10";
        while (true) {
            long startTimestamp = System.currentTimeMillis();
            //结束时间等于上次开始时间加10秒
            //开始时间等于上次开始时间
            System.out.println("开始时间：" + startTime + " 结束时间：" + endTime);
            String sql = String.format(dorisQuerySql, startTime, endTime);
            int total = executeSql(dorisUrl, dorisUsername, dorisPassword, sql);
            System.out.println(sql);
            System.out.println("开始时间：" + DateUtils.addDays(DateUtils.parseDate(startTime, "yyyy-MM-dd HH:mm:ss"), ADD_DAY).toLocaleString() +
                    "结束时间：" + DateUtils.addDays(DateUtils.parseDate(endTime, "yyyy-MM-dd HH:mm:ss"), ADD_DAY).toLocaleString() + " 总数：" + total
                    + ",耗时：" + (System.currentTimeMillis() - startTimestamp) + "ms");
            startTime = endTime;
            endTime = addSeconds(startTime, 10);
            long endTimestamp = System.currentTimeMillis();
            if (endTimestamp - startTimestamp < 10000) {
                Thread.sleep(10000 - (endTimestamp - startTimestamp));
            } else {
                System.out.println("耗时：" + (endTimestamp - startTimestamp) + "ms");
            }
        }
    }

    //把日期时间字符串转为时间戳
    private static long getTimestamp(String dateTime) {
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            Date date = dateFormat.parse(dateTime);
            return date.getTime();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return -1;
    }

    //日期时间加上指定的秒数
    private static String addSeconds(String dateTime, int seconds) {
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        try {
            Date date = dateFormat.parse(dateTime);
            long timestamp = date.getTime();
            timestamp += seconds * 1000;
            date = new Date(timestamp);
            return dateFormat.format(date);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    private static int executeSql(String dorisUrl, String dorisUsername, String dorisPassword, String sql) {
        int cnt = 0;
        try (Connection connection = DriverManager.getConnection(dorisUrl, dorisUsername, dorisPassword);
             PreparedStatement preparedStatement = connection.prepareStatement(sql);
             ResultSet resultSet = preparedStatement.executeQuery()) {

            ResultSetMetaData metaData = resultSet.getMetaData();
            int columnCount = metaData.getColumnCount();
            while (resultSet.next()) {
                // 构建整行数据的JSON对象
                Map<String, Object> rowData = new HashMap<>();
                for (int i = 1; i <= columnCount; i++) {
                    String columnName = metaData.getColumnName(i);
                    //把columnName转为驼峰命名,下划线作为一个单词的开始要大写，整个字段名的首字母小写
                    columnName = toCamelCase(columnName);
                    if ("passTime".equals(columnName)) {
                        long timestamp = getTimestamp(resultSet.getString(i));
                        //把时间加上3天
                        timestamp = DateUtils.addDays(new Date(timestamp), ADD_DAY).getTime();
                        rowData.put(columnName, timestamp);
                    } else {
                        rowData.put(columnName, resultSet.getObject(i));
                    }
                }
                // 转换为JSON字符串
                String jsonData = objectMapper.writeValueAsString(rowData);
                // 发送数据到Kafka
                sendVehicleData(jsonData);
                cnt++;
                // System.out.println("已发送数据：" + jsonData);
            }
            // System.out.println("数据发送完毕");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
        }
        return cnt;
    }

    /**
     * 驼峰式命名法
     * 例如：user_name->userName
     *
     * @param s 要转换的字符串
     * @return 转换后的驼峰式命名字符串
     */
    public static String toCamelCase(String s) {
        if (s == null) {
            return null;
        }
        char SEPARATOR = '_';
        if (s.indexOf(SEPARATOR) == -1) {
            return s;
        }
        s = s.toLowerCase();
        StringBuilder sb = new StringBuilder(s.length());
        boolean upperCase = false;
        for (int i = 0; i < s.length(); i++) {
            char c = s.charAt(i);

            if (c == SEPARATOR) {
                upperCase = true;
            } else if (upperCase) {
                sb.append(Character.toUpperCase(c));
                upperCase = false;
            } else {
                sb.append(c);
            }
        }
        return sb.toString();
    }

    private static Properties getKafkaProperties() {
        // 配置 Kafka 生产者
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    private static void sendVehicleData(String jsonData) {
        ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, jsonData);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                System.err.println("发送失败: " + exception.getMessage());
            } else {
                //System.out.println("发送成功: " + metadata.toString());
            }
        });
    }


}